package cc.rboot.topic.processor;

import cc.rboot._core.holder.AppContextHolder;
import cc.rboot.topic.core.AbstractMessageProcessor;
import cc.rboot.topic.core.IMessageConsumer;
import org.redisson.api.RQueue;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;

import java.util.Set;

public class QueueMessageProcessor<T> extends AbstractMessageProcessor<T> {

    @Override
    public void register(IMessageConsumer<T> consumer) {
        boundConsumer(consumer.topic().name()).add(consumer.name());

        new Thread(() -> {
            while (true) {
                try {
                    T message = this.poll(consumer);
                    if (message == null) {
                        continue;
                    }
                    boolean ifSuccess = false;
                    try {
                        ifSuccess = consumer.onMessage(message);
                    } catch (Exception ex) {
                        // Do nothing
                    }
                    if (!ifSuccess) {
                        try {
                            this.sendToConsumer(consumer.name(), message);
                        } catch (Exception ex) {
                            // 打印下日志即可，一般消费者都是回调，出错了也没有报错的对象
                            System.out.println(ex.getMessage());
                        }
                    }
                    continue;
                } catch (Exception ex) {
                    System.out.println("发生错误" + ex.getMessage());
                }
                this.sleep();
            }
        }).start();
    }


    private T poll(IMessageConsumer<T> consumer) throws Exception {
        return boundQueue(consumer.name()).poll();
    }

    @Override
    public void send(T message) throws Exception {
        for (String consumerName : boundConsumerSet()) {
            this.sendToConsumer(consumerName, message);
        }
    }

    public void sendToConsumer(String consumerName, T message) throws Exception {
        boundQueue(consumerName).add(message);
    }

    /**
     * 单轮消费间隔时间
     */
    private void sleep() {
        try {
            Thread.sleep(20);
        } catch (InterruptedException e) {
            // Do nothing
        }
    }

    private Set<String> boundConsumerSet() {
        return boundConsumer(super.getTopic()).readAll();
    }

    private RQueue<T> boundQueue(String consumerName) {
        String key = "MESSAGE:" + this.getTopic() + ":" + consumerName;
        return getClient().getQueue(key);
    }

    private RSet<String> boundConsumer(String topicName) {
        return getClient().getSet("TOPIC:" + topicName);
    }

    private RedissonClient getClient() {
        return ((RedissonClient) AppContextHolder.getBean("redissonClient"));
    }
}
